package com.mjf.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mjf.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 消费 ods 层日志数据（ods_base_log）写入 dwd 层（dwd_start_log/dwd_page_log/dwd_display_log）
 *
 * 数据流向：web/app -> nginx -> springboot -> kafka(ods) -> flinkApp -> kafka(dwd)
 * 程序：gmall2020-mock-log-2020-12-18.jar -> flink-logger.sh(包含 nginx/springboot/kafka(ods)) -> BaseLogApp(包含 flinkApp/kafka(dwd))
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {

        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

/*
        // 测试时关闭
        // 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:9000/gmall-flink/checkpoint"));
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);

        // 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10));
*/
        // 2.消费 ods_base_log 主题数据创建流，数据格式参考 resources/app.log
        String topic = "ods_base_log";
        String groupId = "BaseLogApp";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, groupId));

        // 3.将每行数据转换为 Json 对象
        // 用来存放解析异常的脏数据
        OutputTag<String> dirtyDataOutput = new OutputTag<String>("dirtyData"){};

        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    // 解析异常时，将数据写入侧输出流
                    ctx.output(dirtyDataOutput, value);
                }
            }
        });

        // 4.新老用户校验 状态编程
        SingleOutputStreamOperator<JSONObject> jsonObjWithNewFlag = jsonObjDS
                .keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"))
                .map(new RichMapFunction<JSONObject, JSONObject>() {

                    private ValueState<String> isNewState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        isNewState = getRuntimeContext().getState(new ValueStateDescriptor<String>("is-new", String.class));
                    }

                    @Override
                    public JSONObject map(JSONObject value) throws Exception {
                        // 获取数据中的 is_new 字段
                        String isNew = value.getJSONObject("common").getString("is_new");

                        // 判断 isNew 是否为 1，为1时进行校验，为0时直接输出
                        if ("1".equals(isNew)) {
                            if (isNewState.value() != null) {
                                // 数据中是新用户，且状态中有值时，修改 is_new 字段的值为0
                                value.getJSONObject("common").put("is_new", 0);
                            } else {
                                isNewState.update("1"); // 修改状态值，保证下一次判断时不为 null
                            }
                        }

                        return value;
                    }

                });

        // 5.分流 侧输出流    页面(page)：主流 启动(start)：侧输出流 曝光(displays)：侧输出流
        // 用来存放启动日志
        OutputTag<String> startOutput = new OutputTag<String>("start"){};
        // 用来存放曝光数据
        OutputTag<String> displayOutput = new OutputTag<String>("display"){};
        SingleOutputStreamOperator<String> pageDS = jsonObjWithNewFlag.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                // 获取启动日志字段
                String start = value.getString("start");
                if (start != null && start.length() > 0) {
                    // 将数据写入到启动日志侧输出流
                    ctx.output(startOutput, value.toJSONString());
                } else {
                    // 将数据写入页面日志主流
                    out.collect(value.toJSONString());

                    // 取出数据中的曝光数据(所有的曝光数据都属于页面数据)
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null && displays.size() > 0) {
                        // 获取页面ID
                        String pageId = value.getJSONObject("page").getString("page_id");
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            // 添加 pageId
                            display.put("page_id", pageId);
                            // 将数据写出到曝光侧输出流
                            ctx.output(displayOutput, display.toJSONString());
                        }
                    }
                }

            }
        });

        // 6.提取侧输出流
        DataStream<String> startDS = pageDS.getSideOutput(startOutput);
        DataStream<String> displayDS = pageDS.getSideOutput(displayOutput);
        DataStream<String> dirtyData = jsonObjDS.getSideOutput(dirtyDataOutput);

        // 7.将三个流进行打印并输出到对应的 Kafka 主题中
        dirtyData.print("DirtyData=======>");   // 脏数据
        startDS.print("Start=======>");
        pageDS.print("Page=======>");
        displayDS.print("Display=======>");

        startDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_start_log"));
        pageDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_page_log"));
        displayDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_display_log"));

        // 8.启动任务
        env.execute(BaseLogApp.class.getName());

    }
}
